#include <config.h>
#include <core.h>

static void msgd_drain_connection(void)
{
   msgd_uint_t i;

   for(i=0; i<32; i++){
      if(queue_empty(&global.reusable_connections_queue)){
         break;
      } 

      queue_t *q = queue_last(&global.reusable_connections_queue);
      msgd_connection_t *c = queue_data(q,msgd_connection_t,queue);
     
      DEBUG("reusing connection %p",c);
      c->close = 1;
      c->read->handler(c->read);
   }
}

//msgd_int_t msgd_open_listening_socket(const char *host,const char *port)
msgd_int_t msgd_open_listening_socket(msgd_listening_t *ls)
{
   int sockfd = -1 , tries , failed = 0 , err , reuseaddr;
   struct sockaddr_in *serv_addr = (struct sockaddr_in*)ls->sockaddr;

   for( tries=0; tries<5; tries++){
  
      if((sockfd=socket(AF_INET,SOCK_STREAM,0)) == -1){
          ERROR("server: cant open stream socket");
          return MSGD_ERROR;
      }
     
      if(sockfd == -1){
        ERROR("socket failed sockfd:%d",sockfd);
        return MSGD_ERROR;
      }
   
      reuseaddr = 1;
      if(setsockopt(sockfd,SOL_SOCKET,SO_REUSEADDR,(const void*)&reuseaddr,sizeof(int)) == -1){
         ERROR("setsockopt(SO_REUSEADDR) host:%s port:%s failed",ls->addr_text,ls->port_text);
         if(close(sockfd) == -1)
            ERROR("colse(%d) failed",sockfd);
         return MSGD_ERROR;
      }
   
      if(msgd_setnonblocking(sockfd) == -1 ){
         ERROR("msgd_setnonblocking(%d) failed",sockfd);
         if(close(sockfd) == -1)
            ERROR("colse(%d) failed",sockfd);
         return MSGD_ERROR;
      }
   
      DEBUG("bind %s #%d",ls->port_text,sockfd);
   
      if(bind(sockfd,(struct sockaddr *)serv_addr,ls->socklen) == -1){
         ERROR("server: cant bind local address");
         err = errno; 
         if(close(sockfd) == -1)
            ERROR("colse(%d) failed",sockfd);
         if(err != EADDRINUSE) 
            return MSGD_ERROR;
         failed = 1;
      }else if(listen(sockfd,ls->backlog) == -1){
         ERROR("server: cant listen");
         if(close(sockfd) == -1)
            ERROR("colse(%d) failed",sockfd);
         return MSGD_ERROR;
      }
   
      if(!failed)
        break;
    
      usleep(500 * 1000); //500 ms 
   }

   if(failed){
      ERROR("still could not bind()");
      return MSGD_ERROR;
   }

   ls->fd = sockfd;

   return MSGD_OK;
}


int msgd_setnonblocking(int fd)
{
   int opts;
   opts = fcntl(fd,F_GETFL);
   if(opts == -1){
      ERROR("fcntl F_GETFL %d:error",fd);
      return opts;
   }

   opts |= O_NONBLOCK; 
  
   return fcntl(fd,F_SETFL,opts);
}


void msgd_close_connection(msgd_connection_t *c)
{

   int fd;
   if(c->fd == -1){
     ERROR("%p connections has arealy closed");
     return;
   }

   
   if(c->read->active || c->read->disable){
      msgd_del_event(c->read,MSGD_READ_EVENT,MSGD_CLOSE_EVENT);
   }

   if(c->write->active || c->write->disable){
     msgd_del_event(c->write,MSGD_WRITE_EVENT,MSGD_CLOSE_EVENT);
   }

   fd = c->fd;
   c->fd = -1; 

   msgd_reusable_connection(c,0);
  
   msgd_free_connection(c);

   if(close(fd) == -1){
      ERROR("close connection(%p) failed");
   } 
}


void msgd_close_accepted_connection(msgd_connection_t *c)
{
   if(close(c->fd) == -1){
      ERROR("close(%d) failed",c->fd);
   }

   msgd_free_connection(c); 

   if(c->pool)
     destroy_pool(c->pool); 
}

#if 0
msgd_int_t msgd_connection_error(msgd_connection_t *c, int err, char *text)
{
}
#endif

msgd_connection_t *msgd_get_connection(int s)
{
   msgd_connection_t  *c = global.free_connections;
   if(c == NULL){
     msgd_drain_connection();
     c = global.free_connections; 
   }

   if(c == NULL){
     ERROR("%u worker_connections are not enough",global.connections_n);
     return NULL;
   }

   global.free_connections = c->data;
   global.free_connections_n--;


   msgd_event_t *rev = c->read;
   msgd_event_t *wev = c->write;
   
   memset(c,0,sizeof(msgd_connection_t));
   c->read = rev;
   c->write = wev;
   c->fd = s;

   memset(rev,0,sizeof(msgd_event_t));
   memset(wev,0,sizeof(msgd_event_t));
   
   rev->data = c;
   wev->data = c; 
   wev->write = 1;

   return c;
}

void msgd_free_connection(msgd_connection_t *c)
{
   DEBUG("free connections %p",c);

   c->data = global.free_connections;
   global.free_connections = c;
   global.free_connections_n++;
}

void msgd_reusable_connection(msgd_connection_t *c, msgd_uint_t reusable)
{
   if(c->reusable){
     queue_remove(&c->queue);
   }

   c->reusable = reusable;
   
   if(reusable){
     queue_insert_head((queue_t*)&global.reusable_connections_queue,&c->queue);
   }
}


void msgd_init_connection(msgd_connection_t *c)
{
   INFO("new connection  fd:%d",c->fd);
   msgd_event_t   *rev;
  
   rev = c->read;  
  
   rev->handler = msgd_read_xhead_request_line;
   c->write->handler = msgd_empty_write_handler;
 
   if(msgd_handle_read_event(rev) == MSGD_ERROR){
      msgd_close_connection(c);
   }
}
